package com.bancomer.batch.impl;

import java.io.IOException;
import java.util.List;

import org.apache.log4j.Logger;

import com.bancomer.batch.BatchController;
import com.bancomer.batch.Constants;
import com.bancomer.batch.pig.Pig;
import com.bancomer.batch.pojo.Alert;
import com.bancomer.batch.pojo.FileHelper;
import com.bancomer.batch.pojo.ValidationResult;
import com.bancomer.batch.transferes.DownloadHdfs;
import com.bancomer.batch.transferes.UploadHdfs;
import com.bancomer.batch.util.LinuxCommand;
import com.bancomer.batch.validators.GeneralValidator;
import com.bancomer.batch.validators.impl.ExistingFilesValidator;
import com.bancomer.batch.validators.impl.LayoutValidator;
import com.bancomer.batch.validators.impl.SensitiveCollectiveValidator;


public class BatchCotrollerImpl extends BatchControllerSupport  {

	@Override
	public void init() {
		System.out.println("Init");
	}

	@Override
	public void validateFiles() {
		Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
		// TODO Auto-generated method stub
		List<FileHelper> propertiesFiles = alert.getPropertiesFiles();
		List<FileHelper> hdfsFiles = alert.getHdfsFiles();
		ValidationResult result;
		for(FileHelper propertiesFile:propertiesFiles){
			GeneralValidator existingFileValidator = new ExistingFilesValidator();
			result = existingFileValidator.executeValidation(propertiesFile, null);
			if(result.getSucces()==false){
				super.getBatchContext().setMsgError(result.getMessage());
				super.getBatchContext().setContinueFlow(result.getSucces());
				return;
			}
			if(propertiesFile.getClassValidation()!=null && !propertiesFile.getClassValidation().trim().equals("")){
				Class<GeneralValidator> userClass;
				try {
					userClass = (Class<GeneralValidator>) Class.forName(propertiesFile.getClassValidation());
					GeneralValidator validator= (GeneralValidator) userClass.newInstance();
					result = validator.executeValidation(propertiesFile, alert);
					super.getBatchContext().setContinueFlow(result.getSucces());
					if(result.getSucces()==false){
						super.getBatchContext().setMsgError(result.getMessage());
						super.getBatchContext().setContinueFlow(result.getSucces());
						return;
					}
				} 
				catch (Exception e) {
					super.getBatchContext().setMsgError(e.getMessage());
					super.getBatchContext().setContinueFlow(false);
					return;
				}      
			}
		}
		for(FileHelper fileHdfs:hdfsFiles){
			GeneralValidator existingFileValidator = new ExistingFilesValidator();
			result = existingFileValidator.executeValidation(fileHdfs, null);
			if(result.getSucces()==false){
				super.getBatchContext().setMsgError(result.getMessage());
				super.getBatchContext().setContinueFlow(result.getSucces());
				return;
			}
		}
	}



	@Override
	public void putFilesToHdfs() {
		// TODO Auto-generated method stub
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			List<FileHelper> hdfsFiles = alert.getHdfsFiles();
			for (FileHelper fileHdfs:hdfsFiles){
				UploadHdfs.uploadFilesHdfs(fileHdfs);
			}
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}		
	}

	@Override
	public void executePig() {
		// TODO Auto-generated method stub
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			Pig.executePigScript(alert);
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}	
	}

	@Override
	public void getFilesOnHdfs() {
		// TODO Auto-generated method stub
		try{
			Alert alert = (Alert)super.getBatchContext().getContextParam().get("alert");
			List<FileHelper> outputFiles = alert.getPigScript().getOutputResultsFiles();
			for (FileHelper output:outputFiles){
				DownloadHdfs.downloadFilesHdfs(output);
			}
		}
		catch(Exception e){
			super.getBatchContext().setMsgError(e.getMessage());
			super.getBatchContext().setContinueFlow(false);
			return;
		}	
	}



	@Override
	public void prepareResponseFiles() {
		// TODO Auto-generated method stub
		
	}

	
	

}
